package edu.xhu.kafkaproducer.listener;

import cn.hutool.json.JSON;
import cn.hutool.json.JSONUtil;
import edu.xhu.kafkaproducer.dtos.MessageDto;
import edu.xhu.kafkaproducer.enums.KafkaConsumerGroup;
import edu.xhu.kafkaproducer.enums.KafkaTopic;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import java.util.List;

@Component
@Slf4j
public class KafkaConsumerComponent {
    @KafkaListener(topics = KafkaTopic.BILL_TEST_TOPIC, groupId = KafkaConsumerGroup.BILL_TEST_GROUP)
    public void topic_test(List<String> messages,
                           @Header(KafkaHeaders.RECEIVED_PARTITION_ID) List<Integer> partitions,
                           @Header(KafkaHeaders.RECEIVED_TOPIC) List<String> topics) {
        for (int i = 0; i < messages.size(); i++) {
            String message = messages.get(i);
            Integer partition = partitions.get(i);
            String topic = topics.get(i);
            System.out.println("消费组：" + KafkaConsumerGroup.BILL_TEST_GROUP + " 消费了主题：" + topic + " 的分区：" + partition + " 中的消息：" + message);
            //ack.acknowledge();
        }
        System.out.println("分区:" + partitions);
        System.out.println("主题:" + topics);
        System.out.println();
    }
}
